package com.violet.rabbitmq.config;


import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Correlation;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

import static com.violet.rabbitmq.config.PrimaryRabbitMqConfig.PRIMARY_TEMPLATE;


/**
 * 类说明: mq消息回调配置
 *
 * @author wqf
 * @date 2024/3/19 10:34
 */
@Component
@Slf4j
public class CallBack implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback {

    @Resource(name = PRIMARY_TEMPLATE)
    private RabbitTemplate rabbitTemplate;

    //相当于 注入，在其他注入结束后才进行注入
    @PostConstruct
    private void init() {
        //注入
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnsCallback(this);
    }

    /**
     * 交换机回调方法(针对于交换机是否成功接收消息)
     * 1.发消息 交换机接收到了 回调
     * 1.1 correlationData 保存回调消息的id及相关信息
     * 1.2 交换机收到消息 ack=true
     * 1.3 call null
     * 2. 发消息 交换机失败了 回调
     * 2.1 correlationData 保存回调消息的id及相关信息
     * 2.2 交换机收到消息 ack=false
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        String id = correlationData != null ? correlationData.getId() : null;
        if (ack) {
            log.info("ack,消息投递到exchange成功,msgId:{}", id);
        } else {
            log.error("ack,消息投递exchange失败，msgId:{},原因{}", id, cause);
//            if (correlationData instanceof CorrelationDataExt && StrUtil.isNotBlank(id)) {
//                CorrelationDataExt correlationDataExt = (CorrelationDataExt) correlationData;
//                String message = correlationDataExt.getData();
//                String exchange = ((CorrelationDataExt) correlationData).getExchange();
//                String routingKey = ((CorrelationDataExt) correlationData).getRoutingKey();
//                //记录失败消息的发送次数
//                MqMessageSendLogEntity messageSendLog = new MqMessageSendLogEntity();
//                messageSendLog.setId(Long.valueOf(correlationData.getId()));
//                messageSendLog.setDate(LocalDateTime.now());
//                messageSendLog.setExchange(exchange);
//                messageSendLog.setRoutingKey(routingKey);
//                messageSendLog.setPayload(message);
//                messageSendLogDao.insert(messageSendLog);
//            } else {
//                log.info("correlationData对象不包含数据");
//            }
        }

    }

    /**
     * 交换机返回消息的方法-消息未送达队列触发回调
     * （1）常用于交换机无法路由回退消息。
     * （2）如果交换机绑定了备用交换机则是路由到备用交换机，此方法不回调。
     * （3）如果是发送到延迟交换机则回调此方法，所以如果使用延迟交换机则要对延迟交换机回调的消息过滤。
     *
     * @param returned the returned message and metadata.
     */
    @Override
    public void returnedMessage(ReturnedMessage returned) {
        log.warn("返回消息回调:{} 应答代码:{} 回复文本:{} 交换器:{} 路由键:{}", returned.getMessage(), returned.getReplyCode(),
                returned.getReplyText(), returned.getExchange(), returned.getRoutingKey());
        //记录失败消息的发送次数
        Message message = returned.getMessage();
        String correlationId = message.getMessageProperties().getCorrelationId();
        String data = new String(message.getBody());
//        if (correlationId != null) {
//            //记录失败消息的发送次数
//            MqMessageSendLogEntity messageSendLog = new MqMessageSendLogEntity();
//            messageSendLog.setId(Long.valueOf(correlationId));
//            messageSendLog.setDate(LocalDateTime.now());
//            messageSendLog.setExchange(returned.getExchange());
//            messageSendLog.setRoutingKey(returned.getRoutingKey());
//            messageSendLog.setPayload(data);
//            messageSendLogDao.insert(messageSendLog);
//        }
    }

    @Bean
    public MessagePostProcessor correlationIdProcessor() {
        return new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message, Correlation correlation) {
                MessageProperties messageProperties = message.getMessageProperties();
                if (correlation instanceof CorrelationData) {
                    String correlationId = ((CorrelationData) correlation).getId();
                    messageProperties.setCorrelationId(correlationId);
                }
                return message;
            }
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                return message;
            }
        };
    }
}